package com.cml.infrastructure;

import com.cml.common.config.Configuration;
import com.cml.common.constant.NodeStatus;
import com.cml.common.context.ServerContext;
import com.cml.domain.builder.ServerNodeBuilder;
import com.cml.domain.entity.Message;
import com.cml.domain.entity.ServerNode;
import com.cml.domain.factory.MessageBuilderFactory;
import com.cml.domain.factory.RemoteServerNodeMessageBuilderFactory;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.OutputStream;
import java.net.*;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 姓名：陈茂林
 * 时间：2023/10/7 9:15
 * 描述：
 */
public class NetworkManager {

    private Configuration configuration = Configuration.getInstance();
    private RemoteServerNodeManager remoteServerNodeManager;

    private Map<Integer,Socket> remoteSocket = new ConcurrentHashMap();


    private Map<Integer,IOThreadRunningSignal> ioThreadRunningSignalMap = new ConcurrentHashMap();


    private ServerContext serverContext;



    /**
     * 默认的主动连接的重试次数
     */
    private static final int DEFAULT_CONNECT_RETRIES = 3;


    private static final int DEFAULT_CONNECT_RETRY_THREAD_TIME=30000;
    /**
     * 检查跟其他所有节点的连接状态的时间间隔
     */
    private static final long CHECK_ALL_OTHER_NODES_CONNECT_INTERVAL = 10 * 1000;



    Logger logger = Logger.getLogger(NetworkManager.class);

    public NetworkManager(ServerContext serverContext){
        this.remoteServerNodeManager = new RemoteServerNodeManager();
        this.serverContext = serverContext;
    }
    public void startNetworkConnectionListener(){
        new NetworkConnectionListener().start();
    }

    class NetworkConnectionListener extends  Thread{
        private Configuration configuration = Configuration.getInstance();
        Logger logger = Logger.getLogger(NetworkConnectionListener.class);
        @Override
        public void run() {
            ServerSocket serverSocket =null;
            try {
                SocketAddress endpoint = new InetSocketAddress(configuration.getNodeServerPort());
               serverSocket = new ServerSocket();
               serverSocket.setReuseAddress(true);
                serverSocket.bind(endpoint);
                logger.info("当前启动端口为：" + configuration.getNodeServerPort());
                while (true){
                    Socket socket = serverSocket.accept();
                    socket.setTcpNoDelay(true); // 网络通信不允许延迟
                    socket.setSoTimeout(0); // 读取数据时的超时时间为0，没有超时，阻塞读取

                    if (builderRemoteServerNode(socket)) {
                        continue;
                    }
                    sendCurServerNodeMsg(socket);
                }
            } catch (Exception e) {
                logger.error(e);
                NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
            }finally {
                if(serverSocket != null){
                    try {
                        serverSocket.close();
                    } catch (IOException e) {
                        logger.error(e);
                        NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
                    }
                }
            }

        }


    }

    private boolean builderRemoteServerNode(Socket socket) throws IOException {

        ServerNode remoteServerNode = (ServerNode) MessageBuilderFactory.buildMessage(socket);
        if(remoteServerNode == null){
            logger.error(socket.getRemoteSocketAddress()+"连接信息错误，请检查后进行连接！");
            socket.close();
            return true;
        }

        logger.info(remoteServerNode.getIp()+":"+remoteServerNode.getPort()+"建立连接！！！！！！！！");

        startServerIOThreads(remoteServerNode, socket);
        addRemoteServerNode(remoteServerNode);
        addRremoteSocket(remoteServerNode, socket);
        return false;
    }

    public void clearIOThread(ServerNode remoteServerNode){
        ioThreadRunningSignalMap.remove(remoteServerNode.getId());
    }

    private void startServerIOThreads(ServerNode remoteServerNode,Socket socket) throws IOException {
        IOThreadRunningSignal ioThreadRunningSignal=new IOThreadRunningSignal(true);
        //ioThreadRunningSignal.setIsRunning(true);
        ioThreadRunningSignalMap.put(remoteServerNode.getId(),ioThreadRunningSignal);
        new ServerReadIoThread(ioThreadRunningSignal,this.serverContext,remoteServerNode,socket,MessageQueueManager.getReceiveQueue()).start();
        new ServerWriteIoThread(ioThreadRunningSignal,this.serverContext,remoteServerNode,socket,MessageQueueManager.createSendQueue(remoteServerNode.getId())).start();
    }

    private void addRremoteSocket(ServerNode remoteServerNode,Socket socket){
        this.remoteSocket.put(remoteServerNode.getId(),socket);
    }

    private void removeRremoteSocket(ServerNode remoteServerNode){
        this.remoteSocket.remove(remoteServerNode.getId());
    }

    private void addRemoteServerNode(ServerNode remoteServerNode){
        this.remoteServerNodeManager.addRemoteServerNode(remoteServerNode);
    }

    private void sendCurServerNodeMsg(Socket socket) throws IOException {
        ServerNode remoteServerNode = new ServerNodeBuilder().build();
        if(serverContext.getServerNode().isLeader()){
            remoteServerNode.setLeader(true);
        }
        RemoteServerNodeMessageBuilderFactory remoteServerNodeMessageBuilderFactory = new RemoteServerNodeMessageBuilderFactory();
        ByteBuffer byteBuffer = remoteServerNodeMessageBuilderFactory.buildByteBuffer(remoteServerNode);
        OutputStream outputStream = socket.getOutputStream();
        outputStream.write(byteBuffer.array());
        outputStream.flush();

    }

    public void connAllFollowerServerNode(){
        List<Map>  nodeFollowerServerList=this.configuration.getNodeFollowerServerList();

        for(Map nodeFollowerServer: nodeFollowerServerList){
            sendSocketConnStateful(nodeFollowerServer);
        }
        logger.info("当前连接节点为："+this.remoteServerNodeManager.getAllRemoteServerNode());
    }

    public Map getAllRemoteServerNode(){
        return this.remoteServerNodeManager.getAllRemoteServerNode();
    }

    public RemoteServerNodeManager getRemoteServerNodeManager(){
        return this.remoteServerNodeManager;
    }

    public void connOtherFollowerServerNode(){
        List<Map> nodeFollowerServerList=nodeFollowerServerList=this.configuration.getBeforeFollowerServerList();
        for(Map nodeFollowerServer: nodeFollowerServerList){
            sendSocketConnStateful(nodeFollowerServer);
        }
    }

    public void waitFollowerServerNodeConn(){
        logger.info("等待所有follower服务连接完成！！");
        //判断当前连接上了的节点数是否完成
        int otherFollowerServerNum = this.configuration.getNodeOtherFollowerServerList().size();
        int allFollowerServerNum=this.remoteServerNodeManager.getAllFollowerServerNode().size();
        boolean isExe = true;
        while (isExe ){
            try {
                //当前节点observer时
                if(otherFollowerServerNum == allFollowerServerNum){
                    isExe=false;
                }
                //当前节点为follower会少一个
                if(otherFollowerServerNum < allFollowerServerNum){
                    isExe=false;
                }
                logger.info("当前连接follower server为："+this.remoteServerNodeManager.getAllRemoteServerNode());
                Thread.sleep(NetworkManager.CHECK_ALL_OTHER_NODES_CONNECT_INTERVAL);
                //实时注册变化的
                allFollowerServerNum=this.remoteServerNodeManager.getAllFollowerServerNode().size();
            } catch (InterruptedException e) {
                logger.error("等待所有follower server服务连接报错！");
                NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
            }
        }

    }

    public void sendAllFollowerServerNodeMessage(Message message) throws Exception {
        List<ServerNode> allFollowerServerNode=this.remoteServerNodeManager.getAllFollowerServerNode();
        for (ServerNode serverNode : allFollowerServerNode){
            if(configuration.getNodeServerId().equals(serverNode.getId())){
                //本节点不发
                continue;
            }
            MessageQueueManager.sendMessage(serverNode.getId(),message);
        }
    }

    public void sendAllServerNodeMessage(Message message) throws Exception {
        Collection<ServerNode> allServerNode=this.remoteServerNodeManager.getAllRemoteServerNode().values();
        for (ServerNode serverNode : allServerNode){
            if(configuration.getNodeServerId().equals(serverNode.getId())){
                //本节点不发
                continue;
            }
            MessageQueueManager.sendMessage(serverNode.getId(),message);
        }
    }

    public void waitAllServerNodeConn(){
        logger.info("等待所有服务连接完成！！");
        //判断当前连接上了的节点数是否完成
        while (!this.configuration.getClusterNodeCount().equals(this.remoteServerNodeManager.getAllRemoteServerNode().size()) ){
            try {
                logger.info("当前连接server为："+this.remoteServerNodeManager.getAllRemoteServerNode());
                Thread.sleep(NetworkManager.CHECK_ALL_OTHER_NODES_CONNECT_INTERVAL);
            } catch (InterruptedException e) {
                logger.error("等待所有服务连接报错！");
                NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
            }
        }
        logger.info("当前连接节点为："+this.remoteServerNodeManager.getAllRemoteServerNode());
    }

    private void sendSocketConnStateful(Map connMsg){
        if(!sendSocketConn(connMsg)){
            NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
        }

    }

    public boolean sendSocketConn(Map connMsg)  {

        Socket socket = null;
        int retryNum = 0;
        while (NodeStatus.getInstance().isRunning() && retryNum <= NetworkManager.DEFAULT_CONNECT_RETRIES){
            try {
                logger.info(connMsg.get("ip")+":"+connMsg.get("port")+"连接第"+(retryNum+1)+"次");
                Thread.sleep(NetworkManager.DEFAULT_CONNECT_RETRY_THREAD_TIME);
                socket = new Socket();
                socket.setTcpNoDelay(true);
                socket.setSoTimeout(0);
                socket.connect(new InetSocketAddress((String) connMsg.get("ip"), Integer.parseInt((String)connMsg.get("port")) ));
                this.sendCurServerNodeMsg(socket);
                this.builderRemoteServerNode(socket);
                logger.info(connMsg.get("ip")+":"+connMsg.get("port")+"连接成功");
                break;
            } catch (IOException e) {
                logger.error(e);
                if(socket != null){
                    try {
                        socket.close();
                    } catch (IOException ex) {
                        logger.error(ex);
                    }
                }
                retryNum++;
            } catch (InterruptedException e) {
                logger.error(e);
            }
        }
        if(retryNum > NetworkManager.DEFAULT_CONNECT_RETRIES){
            logger.error(connMsg.get("ip")+":"+connMsg.get("port")+"连接失败！");
           return false;
        }
        return true;
    }

    public  void clearResource(ServerNode remoteServerNode){
        logger.info("移出节点"+remoteServerNode.getId()+"网络相关资源");
        remoteServerNodeManager.removeRemoteServerNode(remoteServerNode);
        this.removeRremoteSocket(remoteServerNode);
        logger.info("移出节点"+remoteServerNode.getId()+"网络相关资源结束");
    }

}
